package raft

//
// this is an outline of the API that raft must expose to
// the service (or tester). see comments below for
// each of these functions for more details.
// 这是一个raft必须暴露给服务的api概述。阅读以下注释中的函数了解详情。
//
// rf = Make(...)
//   create a new Raft server. 创建一个新的raft server。
// rf.Start(command interface{}) (index, term, isleader)
//   start agreement on a new log entry
// rf.GetState() (term, isLeader)
//   ask a Raft for its current term, and whether it thinks it is leader 询问raft当前term和自己是不是leader
// ApplyMsg
//   each time a new entry is committed to the log, each Raft peer
//   should send an ApplyMsg to the service (or tester)
//   in the same server.
//   每次一个新的entry被提交到log列表，每个raft节点应该发送ApplyMsg给服务。
//

import (
	//	"bytes"

	"bytes"
	"fmt"
	"math/rand"
	"sync"
	"sync/atomic"
	"time"

	//	"6.824/labgob"
	"6.824/labgob"
	"6.824/labrpc"
)

const (
	FOLLOWER  = 0
	CANDIDATE = 1
	LEADER    = 2
)

const (
	LOG1 = "LOG1"
	VOTE = "VOTE"
	PERS = "PERS"
	INIT = "INIT"
	APPD = "APPD"
	TIMR = "TIMR"
	COVT = "COVT"
	COMT = "COMT"
)

const (
	VOTE_FAIL = 0
	VOTE_OK   = 1
	VOTE_ERR  = 2
)

const ElectionTimeout = 250
const CandidateTimeout = 240
const heartsBeatsPeriod = 110

// as each Raft peer becomes aware that successive log entries are
// committed, the peer should send an ApplyMsg to the service (or
// tester) on the same server, via the applyCh passed to Make(). set
// CommandValid to true to indicate that the ApplyMsg contains a newly
// committed log entry.
// 当每个raft节点发觉连续log entries 被提交， 节点应该发送一个ApplyMsg给同一个服务器的服务，
// 通过 applyCh 传递给make(). 设CommandValid = true 表明ApplyMsg包含了一个新的commited log entry.
//
// in part 2D you'll want to send other kinds of messages (e.g.,
// snapshots) on the applyCh, but set CommandValid to false for these
// other uses.
// 在2D部分你会发送其他类型的消息在applyCh, 用于其它用途设CommandValid=false
type ApplyMsg struct {
	CommandValid bool
	Command      interface{}
	CommandIndex int
	CommandTerm  int

	// For 2D:
	SnapshotValid bool
	Snapshot      []byte
	SnapshotTerm  int
	SnapshotIndex int
}

// A Go object implementing a single Raft peer.
// raft节点的实现
type Raft struct {
	mu        sync.Mutex          // Lock to protect shared access to this peer's state 用于保护共享状态的锁
	peers     []*labrpc.ClientEnd // RPC end points of all peers                        rpc终端
	persister *Persister          // Object to hold this peer's persisted state
	me        int                 // this peer's index into peers[]
	dead      int32               // set by Kill()
	role      int32               // follower, condidate, leader

	// Your data here (2A, 2B, 2C).
	// Look at the paper's Figure 2 for a description of what
	// state a Raft server must maintain.

	// custom
	startTime       time.Time
	peersNum        int
	lastHeartsbeats time.Time
	applyCh         chan ApplyMsg
	leaderId        int

	// persistent state on all servers
	currentTerm int
	voteFor     int
	logs        []ApplyMsg

	// volatile state on all servers
	commitIndex  int
	lastAppliyed int
	// volatile satae on leader
	nextIndex  []int
	matchIndex []int
}

// return currentTerm and whether this server
// believes it is the leader.
func (rf *Raft) GetState() (int, bool) {

	var term int
	var isleader bool
	// Your code here (2A).
	rf.mu.Lock()
	defer rf.mu.Unlock()
	term = rf.currentTerm
	isleader = rf.role == LEADER
	return term, isleader
}

// save Raft's persistent state to stable storage,
// where it can later be retrieved after a crash and restart.
// see paper's Figure 2 for a description of what should be persistent.
func (rf *Raft) persist() {
	// Your code here (2C).
	// Example:
	w := new(bytes.Buffer)
	e := labgob.NewEncoder(w)
	e.Encode(rf.voteFor)
	e.Encode(rf.currentTerm)
	e.Encode(rf.logs)
	data := w.Bytes()
	rf.persister.SaveRaftState(data)
}

// restore previously persisted state.
func (rf *Raft) readPersist(data []byte) {
	if data == nil || len(data) < 1 { // bootstrap without any state?
		return
	}
	// Your code here (2C).
	// Example:
	r := bytes.NewBuffer(data)
	d := labgob.NewDecoder(r)
	var voteFor int
	var currentTerm int
	var logs []ApplyMsg
	err1 := d.Decode(&voteFor)
	err2 := d.Decode(&currentTerm)
	err3 := d.Decode(&logs)
	if err1 != nil || err2 != nil || err3 != nil {
		rf.debug(PERS, "read persist failed with err:%v,%v,%v", err1, err2, err3)
	} else {
		rf.currentTerm = currentTerm
		rf.voteFor = voteFor
		rf.logs = logs
		rf.debug(PERS, "%v read persist currentTerm=%v, voteFor=%v", rf.toStr(), currentTerm, voteFor)
		// rf.debug(PERS, "%v read persist currentTerm=%v, voteFor=%v, %v", rf.toStr(), currentTerm, voteFor, rf.logsFormat())
	}
}

// A service wants to switch to snapshot.  Only do so if Raft hasn't
// have more recent info since it communicate the snapshot on applyCh.
func (rf *Raft) CondInstallSnapshot(lastIncludedTerm int, lastIncludedIndex int, snapshot []byte) bool {

	// Your code here (2D).

	return true
}

// the service says it has created a snapshot that has
// all info up to and including index. this means the
// service no longer needs the log through (and including)
// that index. Raft should now trim its log as much as possible.
func (rf *Raft) Snapshot(index int, snapshot []byte) {
	// Your code here (2D).

}

// example RequestVote RPC arguments structure.
// field names must start with capital letters!
type RequestVoteArgs struct {
	// Your data here (2A, 2B).
	PreVote      bool
	Term         int
	CandidateId  int
	LastLogIndex int
	LastLogTerm  int
}

// example RequestVote RPC reply structure.
// field names must start with capital letters!
type RequestVoteReply struct {
	// Your data here (2A).
	Term        int
	VoteGranted bool
}

type AppendEntriesArgs struct {
	Term         int
	LeaderId     int
	PrevLogIndex int
	PrevLogTerm  int
	Entries      []ApplyMsg
	LeaderCommit int
	LastLogIndex int
	LastLogTerm  int
}

type AppendEntriesReply struct {
	Term    int
	Success bool

	NotMatch  bool
	BackIndex int
	BackTerm  int
}

// example RequestVote RPC handler.
func (rf *Raft) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply) {
	rf.mu.Lock()
	defer rf.mu.Unlock()
	term := rf.currentTerm
	// candidate的term小于等于本server，不投票
	voteType := "voted"
	if args.PreVote {
		voteType = "pre voted"
	}

	// 和Figure2不一样， reply false if term < currentTerm
	if rf.currentTerm >= args.Term {
		rf.debug(VOTE, "%v %v false to %d cause thisTerm > candidate Term", rf.toStr(), voteType, args.CandidateId)
		reply.Term = term
		reply.VoteGranted = false
		return
	}

	// leader不给 logs落后的candidate投票
	if rf.checkRole(LEADER) && rf.logsSame(args.LastLogTerm, args.LastLogIndex) {
		rf.debug(VOTE, "%v %v false to %d cause leader don't vote out-of-date candidate", rf.toStr(), voteType, args.CandidateId)
		reply.VoteGranted = false
		reply.Term = term
		return
	}

	// candidate的日志一样长或者比较新都投票
	if rf.logsSame(args.LastLogTerm, args.LastLogIndex) || rf.logsLater(args.LastLogTerm, args.LastLogIndex) {
		if !args.PreVote {
			rf.becomeFollower(args.Term)
			rf.voteFor = args.CandidateId
			rf.persist()
		}
		reply.VoteGranted = true
		reply.Term = term
	} else {
		rf.debug(VOTE, "%v %v to [id:%d,term:%d] false, cause logs least", rf.toStr(), voteType, args.CandidateId, args.Term)
		//follower term小但是logs较新，更新term
		reply.VoteGranted = false
		reply.Term = term
	}
}

func (rf *Raft) AppendEntries(args *AppendEntriesArgs, reply *AppendEntriesReply) {
	rf.mu.Lock()
	defer rf.mu.Unlock()

	if (rf.currentTerm <= args.Term) ||
		(rf.checkRole(CANDIDATE) && rf.logsLater(args.LastLogTerm, args.LastLogIndex)) {
		rf.becomeFollower(args.Term)
		rf.leaderId = args.LeaderId
		// add entries
		if !rf.existLog(args.PrevLogIndex, args.PrevLogTerm) {
			reply.Success = false
			reply.Term = rf.currentTerm

			log := rf.leastFirstLogLeft(args.PrevLogTerm)
			reply.NotMatch = true
			reply.BackTerm = log.CommandTerm
			reply.BackIndex = log.CommandIndex
		} else {
			//todo 不能直接截断，因为可能收到过期的heartsBeats，那就相当于回退了已经从leader收到的log
			//todo 响应丢失，sender会在下一次tick再次replica已经复制了的日志
			for index, e := range args.Entries {
				if rf.existLog(e.CommandIndex, e.CommandTerm) {
					continue
				}
				rf.logs = rf.logs[:e.CommandIndex]
				rf.agreement(args.Entries[index:]...)
				rf.persist()
				break
			}
			// rf.logs = rf.logs[:args.PrevLogIndex+1]
			// rf.agreement(args.Entries...)
			// rf.persist()
			for i := rf.commitIndex + 1; i <= args.LeaderCommit; i++ {
				rf.commit(rf.logs[i])
			}
			reply.Success = true
			reply.Term = rf.currentTerm
			// debug("%v accept a new log:%v, logs:%v", rf.toStr(), formatArgLogs(args), rf.logsFormat())
		}
	} else {
		reply.Success = false
		reply.Term = rf.currentTerm
	}
	rf.debug(LOG1, "%v<-[id:%d,term:%d] log:%v, args.PrevLog=[%v,%v]", rf.toStr(), args.LeaderId,
		args.Term, formatArgLogs(args), args.PrevLogIndex, args.PrevLogTerm)
	// rf.debug(LOG1, "%v<-[id:%d,term:%d] log:%v, logs=%v, args.PrevLog=[%v,%v]", rf.toStr(), args.LeaderId,
	// args.Term, formatArgLogs(args), rf.logsFormat(), args.PrevLogIndex, args.PrevLogTerm)
}

// example code to send a RequestVote RPC to a server.
// server is the index of the target server in rf.peers[].
// expects RPC arguments in args.
// fills in *reply with RPC reply, so caller should
// pass &reply.
// the types of the args and reply passed to Call() must be
// the same as the types of the arguments declared in the
// handler function (including whether they are pointers).
//
// The labrpc package simulates a lossy network, in which servers
// may be unreachable, and in which requests and replies may be lost.
// Call() sends a request and waits for a reply. If a reply arrives
// within a timeout interval, Call() returns true; otherwise
// Call() returns false. Thus Call() may not return for a while.
// A false return can be caused by a dead server, a live server that
// can't be reached, a lost request, or a lost reply.
//
// Call() is guaranteed to return (perhaps after a delay) *except* if the
// handler function on the server side does not return.  Thus there
// is no need to implement your own timeouts around Call().
//
// look at the comments in ../labrpc/labrpc.go for more details.
//
// if you're having trouble getting RPC to work, check that you've
// capitalized all field names in structs passed over RPC, and
// that the caller passes the address of the reply struct with &, not
// the struct itself.
func (rf *Raft) sendRequestVote(server int, args *RequestVoteArgs, reply *RequestVoteReply) bool {
	ok := rf.peers[server].Call("Raft.RequestVote", args, reply)
	return ok
}

func (rf *Raft) sendAppendEntries(server int, args *AppendEntriesArgs, reply *AppendEntriesReply) bool {
	ok := rf.peers[server].Call("Raft.AppendEntries", args, reply)
	return ok
}

//
// the service using Raft (e.g. a k/v server) wants to start
// agreement on the next command to be appended to Raft's log. if this
// server isn't the leader, returns false. otherwise start the
// agreement and return immediately. there is no guarantee that this
// command will ever be committed to the Raft log, since the leader
// may fail or lose an election. even if the Raft instance has been killed,
// this function should return gracefully.
// 这个服务使用Raft 想启动agreement 当下一个命令添加到Raft的log。
// 如果服务器不是leader,返回false。否则启动agreement并立刻返回。
// leader可能在选举中失败或者失联，因此不能保证这一命令永远提交给 raft log。
// 即使Raft实例已经被终止，这个函数也应该优雅地返回。

// the first return value is the index that the command will appear at
// if it's ever committed. the second return value is the current
// term. the third return value is true if this server believes it is
// the leader.
// 第一个返回值是命令提交时将出现的索引。第二个返回值是当前任期。如果此服务器认为自己是领导者，则第三个返回值为true。

// Start(command) 要求 Raft 启动agreement，将命令追加到日志副本中。
// Start()应立即返回，无需等待日志追加完成。
// 该服务希望你将每个新的日志条目，封装为ApplyMsg，发送给Make函数中的applyCh参数（这是一个channel）。
func (rf *Raft) Start(command interface{}) (int, int, bool) {
	rf.mu.Lock()
	defer rf.mu.Unlock()
	index := len(rf.logs)
	currTerm := rf.currentTerm
	isLeader := rf.checkRole(LEADER)

	// Your code here (2B).
	// need lock
	if isLeader {
		msg := ApplyMsg{
			CommandValid: true,
			Command:      command,
			CommandIndex: index,
			CommandTerm:  currTerm,
		}
		rf.agreement(msg)
		rf.nextIndex[rf.me] = index + 1
		rf.matchIndex[rf.me] = index
		rf.persist()
		rf.debug(APPD, "%v append entries: %v", rf.toStr(), command)
		// rf.debug(APPD, "%v append entries, logs:%v", rf.toStr(), rf.logsFormat())
		for i := 0; i < rf.peersNum; i++ {
			if i == rf.me {
				continue
			}
			go rf.replica(APPD, i, false)
		}
	}
	return index, currTerm, isLeader
}

func (rf *Raft) createAppendEntries(peer int) AppendEntriesArgs {
	prevLogIndex := 0
	prevLogTerm := 0
	if rf.nextIndex[peer] > 1 {
		prevLog := rf.logs[rf.nextIndex[peer]-1]
		prevLogIndex = prevLog.CommandIndex
		prevLogTerm = prevLog.CommandTerm
	}

	args := AppendEntriesArgs{
		Term:         rf.currentTerm,
		LeaderId:     rf.me,
		PrevLogIndex: prevLogIndex,
		PrevLogTerm:  prevLogTerm,
		Entries:      make([]ApplyMsg, 0),
		LeaderCommit: rf.commitIndex,
		LastLogTerm:  rf.lastLogTerm(),
		LastLogIndex: rf.lastLogIndex(),
	}
	args.Entries = append(args.Entries, rf.logs[rf.nextIndex[peer]:]...)
	return args
}

func (rf *Raft) replica(logType string, peer int, retry bool) {
	rf.mu.Lock()
	if !rf.checkRole(LEADER) {
		rf.mu.Unlock()
		return
	}
	args := rf.createAppendEntries(peer)
	rf.mu.Unlock()

	reply := AppendEntriesReply{}
	ok := rf.sendAppendEntries(peer, &args, &reply)

	// result
	rf.mu.Lock()
	defer rf.mu.Unlock()
	if !rf.checkRole(LEADER) {
		return
	}

	if !ok {
		if retry {
			go rf.replica(logType, peer, true)
		}
	} else if reply.Term > rf.currentTerm {
		rf.becomeFollower(reply.Term)
	} else if !reply.Success {
		// if rf.nextIndex[peer] > 1 {
		// 	rf.nextIndex[peer] -= 1
		// }

		log := rf.leastFirstLogLeft(reply.BackTerm)
		rf.nextIndex[peer] = log.CommandIndex

		if retry {
			go rf.replica(logType, peer, retry)
		}
	} else {
		if len(args.Entries) > 0 {
			lastLog := args.Entries[len(args.Entries)-1]
			index := lastLog.CommandIndex
			term := lastLog.CommandTerm
			rf.nextIndex[peer] = index + 1
			rf.matchIndex[peer] = index
			// 不会通过计算副本数目的方式来提交之前任期内的日志条目
			if term == rf.currentTerm && rf.commitIndex < index && (rf.nReplica(index) > rf.peersNum/2) {
				for i := rf.commitIndex + 1; i <= index; i++ {
					rf.commit(rf.logs[i])
				}
			}
		}
	}
	rf.debug(logType, "%v replica to server[%d] return %v, result:%v, nextIndex=%v, matchIndex=%v", rf.toStr(), peer, ifThen(ok, "ok", "fail"), reply, rf.nextIndex, rf.matchIndex)
}

// unsafe
func (rf *Raft) nReplica(index int) int {
	count := 0
	for i := 0; i < len(rf.matchIndex); i++ {
		if rf.matchIndex[i] >= index {
			count += 1
		}
	}
	return count
}

func (rf *Raft) agreement(commands ...ApplyMsg) {
	rf.logs = append(rf.logs, commands...)
}

func (rf *Raft) commit(msg ApplyMsg) {
	rf.commitIndex = msg.CommandIndex
	rf.applyCh <- msg
	rf.debug(COMT, "%v commit index=%d", rf.toStr(), msg.CommandIndex)
}

// the tester doesn't halt goroutines created by Raft after each test,
// but it does call the Kill() method. your code can use killed() to
// check whether Kill() has been called. the use of atomic avoids the
// need for a lock.
//
// the issue is that long-running goroutines use memory and may chew
// up CPU time, perhaps causing later tests to fail and generating
// confusing debug output. any goroutine with a long-running loop
// should call killed() to check whether it should stop.
func (rf *Raft) Kill() {
	atomic.StoreInt32(&rf.dead, 1)
	// Your code here, if desired.
}

func (rf *Raft) killed() bool {
	z := atomic.LoadInt32(&rf.dead)
	return z == 1
}

func (rf *Raft) checkRole(role int32) bool {
	z := atomic.LoadInt32(&rf.role)
	return z == role
}

func (rf *Raft) setRole(role int32) {
	atomic.StoreInt32(&rf.role, role)
}

func (rf *Raft) ticker() {
	rand.Seed(makeSeed())
	period := time.Duration(ElectionTimeout+(rand.Int63()%150)) * time.Millisecond
	sleepOnce := period
	for !rf.killed() && rf.checkRole(FOLLOWER) {

		rf.mu.Lock()
		lastHeartbeats := rf.lastHeartsbeats
		rf.mu.Unlock()

		if lastHeartbeats.Add(period).After(time.Now()) {
			sleepOnce = period - (time.Since(lastHeartbeats))
			time.Sleep(sleepOnce)
		} else {
			preVoted := rf.preVote(50 * time.Millisecond)
			if preVoted {
				rf.mu.Lock()
				lastHeartbeats = rf.lastHeartsbeats
				if rf.checkRole(FOLLOWER) && lastHeartbeats.Add(period).Before(time.Now()) {
					rf.debug(TIMR, "%v sleep %dms timeout, start election", rf.toStr(), sleepOnce/time.Millisecond)
					rf.becomeCandidate()
					rf.mu.Unlock()
					return
				} else {
					rf.mu.Unlock()
				}
			}
		}
	}
}

func (rf *Raft) heartsBeatsTicker() {
	for !rf.killed() && rf.checkRole(LEADER) {
		rf.mu.Lock()
		if !rf.checkRole(LEADER) {
			rf.mu.Unlock()
			return
		}
		for i := 0; i < rf.peersNum; i++ {
			if i != rf.me {
				go rf.replica(TIMR, i, false)
			}
		}
		rf.mu.Unlock()
		time.Sleep(heartsBeatsPeriod * time.Millisecond)
	}
}

func (rf *Raft) createRequestVoteArgs(preVote bool) RequestVoteArgs {
	args := RequestVoteArgs{
		PreVote:      preVote,
		Term:         rf.currentTerm,
		CandidateId:  rf.me,
		LastLogIndex: 0,
		LastLogTerm:  0,
	}
	args.LastLogTerm = rf.logs[len(rf.logs)-1].CommandTerm
	args.LastLogIndex = rf.logs[len(rf.logs)-1].CommandIndex
	if preVote {
		args.Term += 1
	}
	return args
}

func (rf *Raft) election() {
	rf.mu.Lock()
	if !rf.checkRole(CANDIDATE) {
		rf.mu.Unlock()
		return
	}
	ms := CandidateTimeout + (rand.Int63() % 150)
	period := time.Duration(ms) * time.Millisecond
	args := rf.createRequestVoteArgs(false)
	rf.mu.Unlock()

	var votesLock sync.Mutex
	votes := make([]int, rf.peersNum)
	votes[rf.me] = VOTE_OK
	latch := newDownLatch(rf.peersNum - 1)
	for i := 0; i < rf.peersNum; i++ {
		if i == rf.me {
			continue
		}
		go func(server int) {
			reply := RequestVoteReply{}
			// DPrintf("server[%v] send vote to %v\n", rf.me, i)
			ok := rf.sendRequestVote(server, &args, &reply)
			rf.mu.Lock()
			rf.debug(VOTE, "%s<-[id:%d,term:%d] vote %v,result:%v", rf.toStr(), server, reply.Term, ifThen(ok, "ok", "fail"), reply)
			rf.mu.Unlock()
			// update term?
			votesLock.Lock()
			if !ok {
				votes[server] = VOTE_ERR
				// retry
			} else if reply.VoteGranted {
				votes[server] = VOTE_OK
			}
			votesLock.Unlock()
			latch.Down()
		}(i)
	}

	//wait timeout or all voted
	latch.WaitTime(period)

	rf.mu.Lock()
	defer rf.mu.Unlock()
	if rf.checkRole(CANDIDATE) {
		votesLock.Lock()
		voted := majority(votes)
		rf.debug(VOTE, "%v get votes %v", rf.toStr(), votes)
		votesLock.Unlock()
		if voted {
			rf.becomeLeader()
			rf.debug(VOTE, "%v become a leader", rf.toStr())
		} else {
			rf.debug(VOTE, "%v election timeout, try election again", rf.toStr())
			rf.becomeCandidate()
		}
	}
}

func (rf *Raft) preVote(ms time.Duration) bool {
	rf.mu.Lock()
	if !rf.checkRole(FOLLOWER) {
		rf.mu.Unlock()
		return false
	}
	args := rf.createRequestVoteArgs(true)
	rf.mu.Unlock()

	var lock sync.Mutex
	votes := make([]int, rf.peersNum)
	votes[rf.me] = VOTE_OK
	latch := newDownLatch(rf.peersNum - 1)
	for i := 0; i < rf.peersNum; i++ {
		if i != rf.me {
			go func(server int) {
				reply := RequestVoteReply{}
				ok := rf.sendRequestVote(server, &args, &reply)
				// debug("%s<-[id:%d,term:%d] vote %v,result:%v", rf.toStr(), server, reply.Term, ifThen(ok, "ok", "fail"), reply)
				lock.Lock()
				if !ok {
					votes[server] = VOTE_ERR
				} else if reply.VoteGranted {
					votes[server] = VOTE_OK
				}
				lock.Unlock()
				latch.Down()
			}(i)
		}
	}
	latch.WaitTime(ms)
	// debug("%v get votes %v", rf.toStr(), votes)
	lock.Lock()
	voted := majority(votes)
	lock.Unlock()
	return voted
}

// the service or tester wants to create a Raft server. the ports
// of all the Raft servers (including this one) are in peers[]. this
// server's port is peers[me]. all the servers' peers[] arrays
// have the same order. persister is a place for this server to
// save its persistent state, and also initially holds the most
// recent saved state, if any. applyCh is a channel on which the
// tester or service expects Raft to send ApplyMsg messages.
// Make() must return quickly, so it should start goroutines
// for any long-running work.
// 这个服务或者测试用例想创建一个Raft服务器，所有的raft服务网络标识都在peers数组里。
// 本服务器标识符是peers[me]。 所有服务器的peers是相同顺序的。
// persister是服务器保存持久化状态的地方，也是保存最近状态的地方。
// applyCh 是测试用例或者服务期望raft发送ApplyMsg的channel。
// Make函数必须快速返回，它应该为任何长时间运行的工作启动goroutines。
func Make(peers []*labrpc.ClientEnd, me int,
	persister *Persister, applyCh chan ApplyMsg) *Raft {
	rf := &Raft{}
	rf.peers = peers
	rf.persister = persister
	rf.me = me
	rf.startTime = time.Now()
	// Your initialization code here (2A, 2B, 2C).
	rf.peersNum = len(peers)
	rf.role = FOLLOWER
	rf.lastHeartsbeats = time.Now()
	rf.currentTerm = 0
	rf.voteFor = -1
	rf.logs = make([]ApplyMsg, 0)
	rf.lastAppliyed = 0
	rf.commitIndex = 0
	rf.applyCh = applyCh
	msg := ApplyMsg{
		CommandValid: false,
		CommandIndex: 0,
		CommandTerm:  0,
	}
	rf.agreement(msg)
	rf.commit(msg)
	// initialize from state persisted before a crash
	// 恢复到崩溃前状态
	rf.readPersist(persister.ReadRaftState())

	// start ticker goroutine to start elections
	go rf.ticker()
	rf.debug(INIT, "init server[%d]", me)
	return rf
}

// unsafe
func (rf *Raft) becomeCandidate() {
	rf.setRole(CANDIDATE)
	rf.currentTerm++
	rf.voteFor = rf.me
	rf.persist()
	go rf.election()
}

// unsafe
func (rf *Raft) becomeLeader() {
	rf.setRole(LEADER)
	rf.voteFor = -1
	rf.persist()
	rf.nextIndex = make([]int, rf.peersNum)
	rf.matchIndex = make([]int, rf.peersNum)
	for i := 0; i < rf.peersNum; i++ {
		rf.nextIndex[i] = rf.logs[len(rf.logs)-1].CommandIndex + 1
	}
	for i := 0; i < rf.peersNum; i++ {
		rf.matchIndex[i] = 0
	}
	go rf.heartsBeatsTicker()
}

func (rf *Raft) becomeFollower(term int) {
	r := rf.role
	rf.setRole(FOLLOWER)
	rf.lastHeartsbeats = time.Now()
	rf.currentTerm = term
	rf.voteFor = -1
	rf.persist()
	if r != FOLLOWER {
		go rf.ticker()
		rf.debug(COVT, "%v change role, [%s]->[FOLLOWER]", rf.toStr(), roleStr(r))
	}

}

func majority(votes []int) bool {
	count := 0
	for i := 0; i < len(votes); i++ {
		if votes[i] == VOTE_OK {
			count++
		}
	}
	if count > len(votes)/2 {
		return true
	}
	return false
}

func (rf *Raft) debug(logType string, format string, a ...interface{}) (n int, err error) {
	ms := time.Now().Sub(rf.startTime)
	s := fmt.Sprintf("%06d %s %s\n", ms.Milliseconds(), logType, format)
	return DPrintf(s, a...)
}

func debug(logType string, format string, a ...interface{}) (n int, err error) {
	s := fmt.Sprintf("%s %s\n", logType, format)
	return DPrintf(s, a...)
}

func (rf *Raft) toStr() string {
	r := ""
	if rf.checkRole(FOLLOWER) {
		r = "F"
	} else if rf.checkRole(CANDIDATE) {
		r = "C"
	} else {
		r = "L"
	}
	return fmt.Sprintf("%s%d[t:%d]", r, rf.me, rf.currentTerm)
}

func roleStr(val int32) string {
	if val == FOLLOWER {
		return "FOLLOWER"
	}
	if val == LEADER {
		return "LEADER"
	}
	if val == CANDIDATE {
		return "CANDIDATE"
	}
	return ""
}

func ifThen(condition bool, trueVal, falseVal interface{}) interface{} {
	if condition {
		return trueVal
	} else {
		return falseVal
	}
}

func (rf *Raft) logsLater(lastLogTerm int, lastLogIndex int) bool {
	return rf.lastLogTerm() < lastLogTerm ||
		(rf.lastLogTerm() == lastLogTerm && (rf.lastLogIndex() < lastLogIndex))
}

func (rf *Raft) logsSame(lastLogTerm int, lastLogIndex int) bool {
	return rf.lastLogTerm() == lastLogTerm && rf.lastLogIndex() == lastLogIndex
}

func (rf *Raft) logsNewer(lastLogTerm int, lastLogIndex int) bool {
	return !rf.logsSame(lastLogTerm, lastLogIndex) && !rf.logsLater(lastLogTerm, lastLogIndex)
}

func (rf *Raft) lastLogTerm() int {
	return rf.logs[len(rf.logs)-1].CommandTerm
}

func (rf *Raft) lastLogIndex() int {
	return rf.logs[len(rf.logs)-1].CommandIndex
}

func (rf *Raft) existLog(index int, term int) bool {
	len := len(rf.logs)
	if index >= len {
		return false
	}
	log := rf.logs[index]
	if log.CommandTerm == term {
		return true
	} else {
		return false
	}
}

func (rf *Raft) leastFirstLogLeft(term int) ApplyMsg {
	len := len(rf.logs)
	i := len - 1
	for ; i >= 0 && rf.logs[i].CommandTerm > term; i-- {
	}
	t := rf.logs[i].CommandTerm
	for ; i >= 0 && rf.logs[i].CommandTerm == t; i-- {
	}
	return rf.logs[i+1]
}

func (rf *Raft) logsFormat() string {
	str := "(index/term)"
	for i := 0; i < len(rf.logs); i++ {
		entry := rf.logs[i]
		// str += fmt.Sprintf("[%d,%d]", entry.CommandIndex, entry.CommandTerm)
		str += fmt.Sprintf("[%d,%d,%v]", entry.CommandIndex, entry.CommandTerm, entry.Command)
	}
	return str
}

func formatArgLogs(args *AppendEntriesArgs) string {
	s := ""
	for i := 0; i < len(args.Entries); i++ {
		entry := args.Entries[i]
		s += fmt.Sprintf("[%d,%d,%v]", entry.CommandIndex, entry.CommandTerm, entry.Command)
	}
	return s
}
